package com.bangcommunity.bbframe.mq.kafka;

import com.bangcommunity.bbframe.common.annotation.QtSpi;
import com.bangcommunity.bbframe.common.utils.lang.StringUtils;
import com.bangcommunity.bbframe.mq.IQtMqFactory;
import com.bangcommunity.bbframe.mq.consumer.IQtMsgProcessor;
import com.bangcommunity.bbframe.mq.kafka.consumer.KafkaConsumerClient;
import com.bangcommunity.bbframe.mq.kafka.producer.KafkaProducerClient;
import com.bangcommunity.bbframe.mq.producer.IQtProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

/**
 * Created by tanghc on 17/1/20.
 */
@QtSpi("kafka")
public class KafkaMqFactory implements IQtMqFactory {
    static Logger logger = LoggerFactory.getLogger(KafkaMqFactory.class);

    @Override
    public IQtProducer builderPruduer() {
        return KafkaProducerClient.getQtProducer();
    }

    @Override
    public Runnable builderConsumer(String rawTopic, IQtMsgProcessor msgProcessor) {
        return builderConsumer(rawTopic, null, msgProcessor);
    }

    @Override
    public Runnable builderConsumer(String projectCode, String appCode, String topic, IQtMsgProcessor msgProcessor) {
        return builderConsumer(projectCode, appCode, topic, null, msgProcessor);
    }

    @Override
    public Runnable builderConsumer(String rawTopic, Properties properties, IQtMsgProcessor msgProcessor) {
        if (StringUtils.isBlank(rawTopic)) {
            throw new RuntimeException("empty topic exception");
        }
        return KafkaConsumerClient.createConsumerTask(rawTopic, msgProcessor, properties);
    }

    @Override
    public Runnable builderConsumer(String projectCode, String appCode, String topic, Properties properties,
            IQtMsgProcessor msgProcessor) {
        String rawTopic = StringUtils.join(new Object[] { projectCode, appCode, topic }, "_");
        return builderConsumer(rawTopic, properties, msgProcessor);
    }
}
